KAFKA-16154: Broker returns offset for LATEST_TIERED_TIMESTAMP by brandboat · Pull Request #16783 · apache/kafka

brandboat · 2024-08-02T17:39:39Z

This pr support EarliestLocalSpec LatestTierSpec in GetOffsetShell, and add integration tests.

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

showuon

Thanks for the PR. Left some comments.

showuon · 2024-08-03T01:24:35Z

+     * Used to retrieve the offset with the local log start offset,
+     * log start offset is the offset of a log above which reads are guaranteed to be served
+     * from the disk of the leader broker, when Tiered Storage is not enabled, it behaves the same
+     * as the earliest timestamp


nit: Used to retrieve ~~the offset with~~ the local log start offset[.]
[Local] log start offset is the offset of a log above which reads are guaranteed to be served from the disk of the leader broker.

Note: When tiered storage is not enabled, it behaves the same as retrieving the earliest timestamp offset.

showuon · 2024-08-03T01:28:54Z

+     * Used to retrieve the offset with the highest offset of data stored in remote storage,
+     * and when Tiered Storage is not enabled, we won't return any offset (i.e. Unknown offset)


nit:

Used to retrieve ~~the offset with~~ the highest offset of data stored in remote storage[.]

Note: When tiered storage is not enabled, we ~~won't return any offset (i.e.~~ [will return] unknown offset.

Thanks for the comments, already fixed them in the latest commit.

showuon

Overall LGTM! We need integration tests for it. We should add some test like DeleteTopicTest using ExpectListOffsetsAction. I expected it should work like this (I didn't try):

// send records to partition 0
                .expectSegmentToBeOffloaded(broker0, topicA, p0, 0, new KeyValueSpec("k0", "v0"))
                .expectSegmentToBeOffloaded(broker0, topicA, p0, 1, new KeyValueSpec("k1", "v1"))
                .expectEarliestLocalOffsetInLogDirectory(topicA, p0, 2L)
                .produce(topicA, p0, new KeyValueSpec("k0", "v0"), new KeyValueSpec("k1", "v1"),
                        new KeyValueSpec("k2", "v2"))
                // new verification:
                 .expectListOffsets(topicA, 0, LatestTierSpec, 0(?)) <-- should create an expected `EpochEntry` instance

Let me know if you have any problem.

showuon · 2024-08-03T02:32:16Z

I mean, we can create a test called listOffsetTest under org.apache.kafka.tiered.storage.integration to test it.

brandboat · 2024-08-03T08:45:37Z

Overall LGTM! We need integration tests for it. We should add some test like DeleteTopicTest using ExpectListOffsetsAction. I expected it should work like this (I didn't try):

Thanks @showuon , I enabled remote storage in GetOffsetShellTest, do we still need the test you mentioned above ?

showuon · 2024-08-03T08:49:03Z


+    private static List<ClusterConfig> withRemoteStorage() {
+        Map<String, String> serverProperties = new HashMap<>();
+        serverProperties.put(RemoteLogManagerConfig.DEFAULT_REMOTE_LOG_METADATA_MANAGER_CONFIG_PREFIX + TopicBasedRemoteLogMetadataManagerConfig.REMOTE_LOG_METADATA_TOPIC_REPLICATION_FACTOR_PROP, "1");


Could we set remote.log.metadata.topic.num.partitions to 1 (default is 50) to speed up the test?

showuon

Overall LGTM! Left some comments. Thanks.

showuon · 2024-08-03T10:53:58Z

+        Map<String, String> zkProperties = new HashMap<>(serverProperties);
+        zkProperties.put(RemoteLogManagerConfig.REMOTE_LOG_METADATA_MANAGER_LISTENER_NAME_PROP, "PLAINTEXT");
+
+        Map<String, String> raftProperties = new HashMap<>(serverProperties);
+        raftProperties.put(RemoteLogManagerConfig.REMOTE_LOG_METADATA_MANAGER_LISTENER_NAME_PROP, "EXTERNAL");


Could we add a comment here to explain why we need 2 different settings?

Correct me if I'm wrong, after spending some time digging through the code I found out that in Type.ZK the default listener name is PLAINTEXT and is set via

kafka/core/src/test/java/kafka/test/junit/ZkClusterInvocationContext.java

Lines 312 to 316 in 96989e4

@Override

public ListenerName listenerName() {

return clusterConfig.listenerName().map(ListenerName::normalised)

.orElseGet(() -> ListenerName.forSecurityProtocol(securityProtocol()));

}

while in Type.KRAFT/Type.CO_KRAFT, it is EXTERNAL, see

kafka/core/src/test/java/kafka/testkit/KafkaClusterTestKit.java

Lines 324 to 332 in 1084d3b

private String listeners(int node) {

if (nodes.isCombined(node)) {

return "EXTERNAL://localhost:0,CONTROLLER://localhost:0";

}

if (nodes.controllerNodes().containsKey(node)) {

return "CONTROLLER://localhost:0";

}

return "EXTERNAL://localhost:0";

}

Another thing is that ClusterConfig#setListenerName only works under ZK mode, in KRAFT it is useless. Perhaps we should make it work under Raft mode too. gentle ping @chia7712, any thoughts ?

@brandboat you are right. Could you please open a jira for that? KRAFT should honor the listener name and security protocol from ClusterConfig.

thanks for double checking, filed https://issues.apache.org/jira/browse/KAFKA-17256

showuon · 2024-08-03T11:09:10Z

+        sendProducerRecords(this::getTopicName);
+    }
+
+    private void setUpRemoteLogTopics() {


I'll add some comment for this method. Ex:

In this method, we'll create 4 topics and produce records to the log like this: topicRLS1 -> 1 segment topicRLS2 -> 2 segments (1 local log segment + 1 segment in the remote storage) topicRLS3 -> 3 segments (1 local log segment + 2 segments in the remote storage) topicRLS4 -> 4 segments (1 local log segment + 3 segments in the remote storage)

Thank you for the detailed comment, this will certainly help others grasp the details of the test 😃

showuon · 2024-08-03T11:12:59Z

+            // test topics enable remote log storage
+            TestUtils.waitForCondition(() ->
+                    Arrays.asList(
+                            new Row("topicRLS2", 0, 0L),
+                            new Row("topicRLS3", 0, 1L),
+                            new Row("topicRLS4", 0, 2L))


I'll add 1 more line for the comment to explain why topicRLS1 has no result:
topicRLS1 has no result because there's no log segments being uploaded to the remote storage

Everything has been addressed in the latest commit, thank you

chia7712

@brandboat thanks for this nice patch. I leave some comments for this PR. The naming is part of public APIs, so we must have "accurate" naming before release.

chia7712 · 2024-08-03T19:33:03Z

+        serverProperties.put(ServerLogConfigs.LOG_INITIAL_TASK_DELAY_MS_CONFIG, "100");
+        serverProperties.put(RemoteLogManagerConfig.REMOTE_LOG_METADATA_MANAGER_LISTENER_NAME_PROP, "EXTERNAL");
+
+        return Arrays.asList(


We can set listener name for all types even though setListenerName is no-op for kraft/co-kraft mode. That can simplify the code and it is still valid.

return Collections.singletonList( ClusterConfig.defaultBuilder() .setTypes(new HashSet<>(Arrays.asList(ZK, KRAFT, CO_KRAFT))) .setServerProperties(serverProperties) .setListenerName("EXTERNAL") .build());

chia7712 · 2024-08-03T19:37:36Z

                return OffsetSpec.latest();
            case "max-timestamp":
                return OffsetSpec.maxTimestamp();
+            case "earliest-local":


Could you please update docs?

kafka/tools/src/main/java/org/apache/kafka/tools/GetOffsetShell.java

Line 135 in 2c9d7af

.describedAs("<timestamp> / -1 or latest / -2 or earliest / -3 or max-timestamp")

chia7712 · 2024-08-03T19:41:41Z

-     * and when Tiered Storage is not enabled, we won't return any offset (i.e. Unknown offset)
+     * Used to retrieve the highest offset of data stored in remote storage.
+     * <br/>
+     * Note: When tiered storage is not enabled, we will return unknown offset.


Could you please add test for this scenario?

this is included in https://github.com/apache/kafka/pull/16783/files/b6426e51d40e5b921f7d497bb0f1c3b2b88adf68#diff-8c7debdc8479f762a5a9585de62d3254c10768249bddd45391b49ba5683526e4R363-R367

chia7712 · 2024-08-03T19:42:04Z

+     * Local log start offset is the offset of a log above which reads
+     * are guaranteed to be served from the disk of the leader broker.
+     * <br/>
+     * Note: When tiered Storage is not enabled, it behaves the same as retrieving the earliest timestamp offset.


Could you please add test for it?

Sure, will do, thanks

chia7712 · 2024-08-03T19:46:29Z

+     * <br/>
+     * Note: When tiered Storage is not enabled, it behaves the same as retrieving the earliest timestamp offset.
     */
    public static OffsetSpec earliestLocalSpec() {


This is unrelated to this PR. The other specs don't use Spec as postfix. Maybe we should align the naming before release?

OK, I'll remove the postfix

chia7712 · 2024-08-03T19:49:23Z

+     * Note: When tiered storage is not enabled, we will return unknown offset.
     */
    public static OffsetSpec latestTierSpec() {
        return new LatestTierSpec();


This is not related to this PR, but KIP-1005 (https://cwiki.apache.org/confluence/display/KAFKA/KIP-1005%3A+Expose+EarliestLocalOffset+and+TieredOffset) calls it LatestTieredSpec rather than LatestTierSpec

ping @FrankYang0529 as you are the author of #16781

Thanks for finding this. Yes, we should follow the KIP. @brandboat, could you fix it in this PR? I'm afraid if I create another PR may block your progress.

Everything has been addressed, thanks everyone.

FrankYang0529 · 2024-08-04T04:51:52Z

@brandboat, I think we also need to update L296-297 in GetOffsetShell

throw new TerseException("Malformed time argument " + listOffsetsTimestamp + ". " +
                            "Please use -1 or latest / -2 or earliest / -3 or max-timestamp, or a specified long format timestamp");

FrankYang0529

LGTM. Thanks for the patch.

showuon · 2024-08-04T07:15:32Z

+        serverProperties.put(RemoteLogManagerConfig.REMOTE_LOG_STORAGE_SYSTEM_ENABLE_PROP, "true");
+        serverProperties.put(RemoteLogManagerConfig.REMOTE_STORAGE_MANAGER_CLASS_NAME_PROP, LocalTieredStorage.class.getName());
+        serverProperties.put(RemoteLogManagerConfig.REMOTE_LOG_MANAGER_TASK_INTERVAL_MS_PROP, "1000");
+        serverProperties.put(TopicBasedRemoteLogMetadataManagerConfig.REMOTE_LOG_METADATA_TOPIC_PARTITIONS_PROP, "1");


I think we should add RemoteLogManagerConfig.DEFAULT_REMOTE_LOG_METADATA_MANAGER_CONFIG_PREFIX to set REMOTE_LOG_METADATA_TOPIC_PARTITIONS_PROP, but I might be wrong. Could you confirm the current change takes effect in the remote log topics?

yeah... I missed that. Thank you

showuon

LGTM!

chia7712

LGTM

brandboat · 2024-08-04T14:44:11Z

failed test passed in my local env:

./gradlew cleanTest \
  core:test --tests kafka.zk.ZkMigrationFailoverTest.testDriverSkipsEventsFromOlderEpoch \
  tool:test --tests org.apache.kafka.tools.LeaderElectionCommandTest.testPreferredReplicaElection \
  clients:test --tests org.apache.kafka.common.record.MemoryRecordsBuilderTest.testBuffersDereferencedOnClose

This pr support EarliestLocalSpec LatestTierSpec in GetOffsetShell, and add integration tests. Reviewers: Luke Chen <showuon@gmail.com>, Chia-Ping Tsai <chia7712@gmail.com>, PoAn Yang <payang@apache.org>

KAFKA-16154: Broker returns offset for LATEST_TIERED_TIMESTAMP

4c4b8b0

showuon reviewed Aug 3, 2024

View reviewed changes

brandboat added 2 commits August 3, 2024 09:38

Address comments

ceabb10

Merge remote-tracking branch 'origin/trunk' into KAFKA-16154

58d38fa

showuon reviewed Aug 3, 2024

View reviewed changes

brandboat added 2 commits August 3, 2024 15:30

Merge remote-tracking branch 'origin/trunk' into test-KAFKA-16154

61db349

Add test

55fb648

Remove debug log

934b8f3

showuon reviewed Aug 3, 2024

View reviewed changes

Comment thread tools/src/test/java/org/apache/kafka/tools/GetOffsetShellTest.java Outdated

showuon reviewed Aug 3, 2024

View reviewed changes

brandboat added 4 commits August 4, 2024 01:04

Address comments

87b5213

Merge remote-tracking branch 'origin/trunk' into KAFKA-16154

c71b83b

Add more comment

7c5d7cc

Modify comment

b6426e5

chia7712 reviewed Aug 3, 2024

View reviewed changes

brandboat added 2 commits August 4, 2024 11:56

Address comments

52f0970

Merge remote-tracking branch 'origin/trunk' into KAFKA-16154

8522e0f

Address comment

d23ef67

FrankYang0529 approved these changes Aug 4, 2024

View reviewed changes

showuon reviewed Aug 4, 2024

View reviewed changes

Add missing config prefix

ce1e59d

showuon approved these changes Aug 4, 2024

View reviewed changes

chia7712 approved these changes Aug 4, 2024

View reviewed changes

showuon merged commit 84add30 into apache:trunk Aug 5, 2024

brandboat deleted the KAFKA-16154 branch August 5, 2024 02:41

		* Used to retrieve the offset with the highest offset of data stored in remote storage,
		* and when Tiered Storage is not enabled, we won't return any offset (i.e. Unknown offset)

	@Override
	public ListenerName listenerName() {
	return clusterConfig.listenerName().map(ListenerName::normalised)
	.orElseGet(() -> ListenerName.forSecurityProtocol(securityProtocol()));
	}

	private String listeners(int node) {
	if (nodes.isCombined(node)) {
	return "EXTERNAL://localhost:0,CONTROLLER://localhost:0";
	}
	if (nodes.controllerNodes().containsKey(node)) {
	return "CONTROLLER://localhost:0";
	}
	return "EXTERNAL://localhost:0";
	}

Uh oh!

Conversation

brandboat commented Aug 2, 2024 • edited by showuon Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Committer Checklist (excluded from commit message)

Uh oh!

showuon left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

showuon left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

showuon commented Aug 3, 2024

Uh oh!

brandboat commented Aug 3, 2024

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

showuon left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

brandboat Aug 3, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

chia7712 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

FrankYang0529 commented Aug 4, 2024

Uh oh!

FrankYang0529 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

brandboat commented Aug 2, 2024 •

edited by showuon

Loading

showuon left a comment •

edited

Loading

brandboat Aug 3, 2024 •

edited

Loading

brandboat commented Aug 4, 2024 •

edited

Loading